package com.arelikebrothers.demo.kafka.stream;

import com.arelikebrothers.demo.kafka.stream.bean.InputMessage;
import com.google.gson.Gson;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.ArrayList;
import java.util.List;
import java.util.Properties;

/**
 * Created by lennon on 16/07/2017.
 */
public class InputProducer {

    private static List<InputMessage> data = new ArrayList<>();

    static {
        data.add(new InputMessage("1", "1", "11"));
        data.add(new InputMessage("1", "1", "111"));
        data.add(new InputMessage("1", "2", "12"));
    }


    public static void main(String[] args) throws Exception {
        Properties config = new Properties();
        config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        config.put(ProducerConfig.CLIENT_ID_CONFIG, "ProducerClientID");
        KafkaProducer<String, String> producer = new KafkaProducer<>(config);
        Gson gson = new Gson();
        while (true)
        {
            for (InputMessage message : data) {
                ProducerRecord<String, String> record = new ProducerRecord<>("INPUT", gson.toJson(message));
                producer.send(record, (RecordMetadata metadata, Exception e)->{
                    if(e != null) {
                        e.printStackTrace();
                        return;
                    }

                    if(metadata != null)
                    {
                        String log = String.format("Send message to Topic:%s, partition:%s, offset:%s, content:%s",
                                metadata.topic(), metadata.partition(), metadata.offset(), record.value());
                        System.out.println(log);

                    }
                });
                Thread.sleep(3000);
            }
        }

    }
}
